import { Channel, Connection, Options } from "amqplib";
import { useMq } from "./useMq";
import { QueueHandler } from "./types";

export async function listenDelayTask(type: string, handler: QueueHandler) {
    let con: Connection | null = null;
    let channel: Channel | null = null;
    try {
        con = await useMq()
        channel = await con.createChannel()
        const topic = `task.topic`
        const waitQueue = `task.wait.${type}`
        const runQueue = `task.run.${type}`
        await channel.assertExchange(topic, 'topic', { durable: true })
        await channel.assertQueue(runQueue, { durable: true, })
        await channel.assertQueue(waitQueue, { durable: true, deadLetterExchange: topic, deadLetterRoutingKey: runQueue })
        await channel.bindQueue(waitQueue, topic, `task.wait.${type}`)
        await channel.bindQueue(runQueue, topic, `task.run.${type}`)
        channel.prefetch(10)
        channel.consume(runQueue, async (msg) => {
            if (msg) await handler(msg, channel!)
        }, { noAck: false })
        return async () => {
            await channel?.cancel(runQueue).catch(e => console.error(e));
            await channel?.close().catch(e => console.error(e))
            await con?.close().catch(e => console.error(e))
        }
    } catch (e) {
        throw e;
    }
    finally {

    }
}

export async function publishToDelayTask(type: string, msg: Buffer, expiration: number, options: Options.Publish = {}) {
    let con: Connection | null = null;
    let channel: Channel | null = null;
    try {
        con = await useMq()
        channel = await con.createChannel()
        return channel.publish(`task.topic`, `task.wait.${type}`, msg, {
            ...options,
            expiration,
        })
    } catch (e) {
        throw e;
    }
    finally {
        if (channel) await channel.close().catch(e => console.error(e))
        if (con) await con.close().catch(e => console.error(e))
    }
}